Please address the comments from this code review:

## Overall Comments
- In ManagedBotWebSocket, `_last_heartbeat_ack_at` is updated but never used to detect missed heartbeats or unhealthy connections; consider either removing it or using it to trigger a forced reconnect on prolonged missing ACKs.
- The new `QQOfficialGatewayUnavailableError` is only raised when `max_concurrency` is missing, but other malformed or incomplete gateway metadata from `get_ws_url` will be accepted; consider tightening validation of the gateway response to avoid starting with unusable state.
- QQOfficialPlatformAdapter.run has been changed to an async method, which may affect existing usage patterns that expect a synchronous `run`; if this adapter is invoked by framework code assuming a sync interface, consider providing a small synchronous wrapper or helper to preserve compatibility.

## Individual Comments

### Comment 1
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py" line_range="112-113" />
<code_context>
+    async def _is_system_event(self, message_event, ws):
+        event_op = message_event["op"]
+        if event_op == self.WS_HELLO:
+            interval_ms = (message_event.get("d") or {}).get("heartbeat_interval")
+            if isinstance(interval_ms, int | float) and interval_ms > 0:
+                self._heartbeat_interval_seconds = interval_ms / 1000
+                logger.info(f"[QQOfficial] Gateway heartbeat interval: {interval_ms}ms")
</code_context>
<issue_to_address>
**issue (bug_risk):** Using `int | float` in `isinstance` is invalid and will raise at runtime.

`isinstance(interval_ms, int | float)` will raise `TypeError` at runtime (`arg 2 must be a type or tuple of types`). Use `isinstance(interval_ms, (int, float))`, or validate/cast the value before use. As currently written, a HELLO with a numeric `heartbeat_interval` will cause `_is_system_event` to raise and may break gateway initialization.
</issue_to_address>

### Comment 2
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py" line_range="195" />
<code_context>
         super().__init__(*args, **kwargs)
         self._shutting_down = False
         self._active_websockets: set[ManagedBotWebSocket] = set()
+        self._next_connect_at = 0.0
+        self._reconnect_attempts = 0
+        self._last_connect_at = 0.0
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the reconnect/backoff and startup retry policies into dedicated helpers so the main client and run loop only interact with a small, high-level API instead of managing scattered state and branching.

You can keep all the behavior but reduce cognitive load by extracting the policy/state into small helpers and collapsing duplicated logic.

### 1. WebSocket reconnect/backoff: extract a policy helper

Right now `botClient` mixes:

- rate-limit handling
- quick-disconnect detection
- exponential backoff
- timestamp tracking (`_next_connect_at`, `_last_connect_at`)

You can encapsulate this in a small object (or even a helper on `botClient`) that owns the state and exposes a simple API.

For example, move the state and logic into a dedicated `ReconnectPolicy`:

```python
class ReconnectPolicy:
    def __init__(self) -> None:
        self._next_connect_at = 0.0
        self._reconnect_attempts = 0
        self._last_connect_at = 0.0
        self._quick_disconnect_count = 0

    def mark_connected(self) -> None:
        self._last_connect_at = time.monotonic()
        self._reconnect_attempts = 0
        self._quick_disconnect_count = 0
        self._next_connect_at = 0.0

    def schedule(
        self,
        *,
        custom_delay: float | None = None,
        rate_limited: bool = False,
    ) -> float:
        delay = custom_delay
        if delay is None and rate_limited:
            delay = _RATE_LIMIT_RECONNECT_DELAY_SECONDS

        if delay is None and self._last_connect_at:
            duration = time.monotonic() - self._last_connect_at
            if duration < _QUICK_DISCONNECT_THRESHOLD_SECONDS:
                self._quick_disconnect_count += 1
            else:
                self._quick_disconnect_count = 0
            if self._quick_disconnect_count >= _MAX_QUICK_DISCONNECTS:
                delay = _RATE_LIMIT_RECONNECT_DELAY_SECONDS
                self._quick_disconnect_count = 0
                logger.warning(
                    "[QQOfficial] Too many quick disconnects; delaying reconnect."
                )

        if delay is None:
            idx = min(
                self._reconnect_attempts,
                len(_RECONNECT_DELAYS_SECONDS) - 1,
            )
            delay = _RECONNECT_DELAYS_SECONDS[idx]
            self._reconnect_attempts += 1

        self._next_connect_at = max(self._next_connect_at, time.monotonic() + delay)
        return delay

    async def wait(self) -> None:
        delay = self._next_connect_at - time.monotonic()
        if delay > 0:
            logger.info(f"[QQOfficial] Waiting {delay:.1f}s before reconnect.")
            await asyncio.sleep(delay)
```

Then `botClient` only forwards:

```python
class botClient(Client):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self._reconnect_policy = ReconnectPolicy()

    def mark_websocket_connected(self) -> None:
        self._reconnect_policy.mark_connected()

    def reset_reconnect_backoff(self) -> None:
        self._reconnect_policy.mark_connected()

    def schedule_reconnect_delay(
        self,
        reason: str,
        *,
        custom_delay: float | None = None,
        rate_limited: bool = False,
    ) -> None:
        delay = self._reconnect_policy.schedule(
            custom_delay=custom_delay,
            rate_limited=rate_limited,
        )
        logger.info(
            f"[QQOfficial] Reconnect scheduled in {delay}s, reason: {reason}"
        )

    async def wait_for_reconnect_delay(self) -> None:
        await self._reconnect_policy.wait()
```

This keeps all existing behavior, but the state is localized and the API from the rest of the code becomes:

- `mark_websocket_connected()`
- `reset_reconnect_backoff()`
- `schedule_reconnect_delay(...)`
- `wait_for_reconnect_delay()`

without `_next_connect_at`, `_reconnect_attempts`, `_quick_disconnect_count` being manipulated directly.

If you prefer not to introduce a class, you can keep the methods on `botClient` but still extract the “quick disconnect” detection into a helper to reduce branching:

```python
def _update_quick_disconnect(self) -> bool:
    if not self._last_connect_at:
        return False
    duration = time.monotonic() - self._last_connect_at
    if duration < _QUICK_DISCONNECT_THRESHOLD_SECONDS:
        self._quick_disconnect_count += 1
    else:
        self._quick_disconnect_count = 0
    return self._quick_disconnect_count >= _MAX_QUICK_DISCONNECTS
```

and then in `schedule_reconnect_delay`:

```python
if delay is None and self._update_quick_disconnect():
    delay = _RATE_LIMIT_RECONNECT_DELAY_SECONDS
    self._quick_disconnect_count = 0
    logger.warning(
        "[QQOfficial] Too many quick disconnects; delaying reconnect."
    )
```

### 2. Startup retry logic: unify error policy + delay

`_should_retry_startup_error` and `_next_startup_retry_delay` both need to know:

- whether the error is retryable
- whether it’s rate-limit-related (string markers `"100017"`, `"频率限制"`, `"Too many requests"`)

You can remove duplication by collapsing them into a single function that returns a delay or `None`:

```python
def _get_startup_retry_delay(self, error: Exception) -> float | None:
    # non-retryable
    if not isinstance(
        error,
        (
            asyncio.TimeoutError,
            ConnectionError,
            OSError,
            QQOfficialGatewayUnavailableError,
            botpy.errors.ServerError,
        ),
    ):
        return None

    delay: float | None = None
    if isinstance(error, botpy.errors.ServerError):
        error_msg = str(error)
        if any(marker in error_msg for marker in ("100017", "频率限制", "Too many requests")):
            delay = _RATE_LIMIT_RECONNECT_DELAY_SECONDS

    if delay is None:
        idx = min(
            self._startup_retry_attempts,
            len(self.STARTUP_RETRY_DELAYS_SECONDS) - 1,
        )
        delay = self.STARTUP_RETRY_DELAYS_SECONDS[idx]
    self._startup_retry_attempts += 1
    return delay
```

Then `run` can be simplified:

```python
async def run(self) -> None:
    while not self._shutdown_event.is_set():
        try:
            await self.client.start(appid=self.appid, secret=self.secret)
            self._startup_retry_attempts = 0
            if self._shutdown_event.is_set():
                break

            logger.warning(
                f"[QQOfficial] Client stopped unexpectedly, restarting in "
                f"{self.STARTUP_RETRY_DELAYS_SECONDS[0]}s."
            )
            await self._restart_client()
            if not await self._sleep_until_retry_or_shutdown(
                self.STARTUP_RETRY_DELAYS_SECONDS[0]
            ):
                break

        except asyncio.CancelledError:
            raise
        except Exception as e:
            if self._shutdown_event.is_set():
                break

            delay = self._get_startup_retry_delay(e)
            if delay is None:
                raise

            logger.warning(
                f"[QQOfficial] Startup failed, retrying in {delay}s: {e}"
            )
            await self._restart_client()
            if not await self._sleep_until_retry_or_shutdown(delay):
                break
```

This keeps:

- same retry classification
- same rate-limit-specific behavior
- same backoff sequence

while removing the duplicated string checks and reducing `run`’s branching.
</issue_to_address>